package com.atguigu.chapter07.c_timer;

import com.atguigu.chapter5.source.WaterSensor;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.time.Duration;

/**
 * @ClassName: Flink14_Timeer_Event
 * @Description:
 * @Author: kele
 * @Date: 2021/4/7 13:16
 *
 * 基于事件时间的定时器
 *
 **/
public class Flink15_Timeer_Event {

    public static void main(String[] args) {

        Configuration conf = new Configuration();
        conf.setInteger("rest.port",20000);

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        env.socketTextStream("hadoop162",8888)
                .map(datas -> {

                    String[] data = datas.split(",");
                    return new WaterSensor(data[0],
                            Long.parseLong(data[1]) * 1000,
                            Integer.valueOf(data[2]));
                })
                //分发水印
                .assignTimestampsAndWatermarks(WatermarkStrategy
                        .<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(3)).withTimestampAssigner(new SerializableTimestampAssigner<WaterSensor>() {

                            //设置时间戳，作为事件的时间
                            @Override
                            public long extractTimestamp(WaterSensor element, long recordTimestamp) {
                                return element.getTs();
                            }
                        }))
                .keyBy(WaterSensor::getId)
                .process(new KeyedProcessFunction<String, WaterSensor, String>() {

                    @Override
                    public void processElement(WaterSensor value, Context ctx, Collector<String> out) throws Exception {

                        ctx.timerService().registerEventTimeTimer(ctx.timestamp()+5000);
                        out.collect(ctx.getCurrentKey()+"注册时间："+value.getTs());
                    }

                    @Override
                    public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {

                        out.collect(ctx.getCurrentKey()+"触发时间："+ timestamp);
                    }
                })
                .print();



        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }


    }

}
